import pandas as pd
import pathway as pw

from application.db.mysql_db.info import (
    ResourceInformationList,
    ResourceSourceDict,
    ResourceInformationSectionList,
    ResourceInformationTagsRelation,
    ResourceInformationAttachmentList,
)
from application.db.mysql_db.nsfc.NsfcInfoList import NsfcInfoList
from application.etl.base_etl import BaseEtl
from application.etl.write.info_to_nsfc_write import NsfcToNsfcWrite
from application.etl.write.print_write import PrintWriter
from application.pipelines.info_process_pipeline import InfoProcessPipeline


class InfoToNsfcPandas(BaseEtl):
    """
    基于 Pandas 的信息消费者 ETL 类。

    功能
    ----
    - 从 MySQL 读取信息表、分段表、标签表、来源字典及 NSFC 信息表。
    - 将分段与标签按 ``information_id`` 聚合为 JSON 列表。
    - 清洗与格式化字段后，生成 pathway 表用于 transform / write。

    特点
    ----
    - 使用 Pandas 中间层便于调试与数据检查。
    - 支持来源校验、反向过滤（anti-join）、时间字段标准化。
    """

    # 可注册自定义 pipelines
    pipelines = [InfoProcessPipeline()]

    def read(self):
        """
        读取源数据并进行聚合与清洗，返回 Pathway 表。

        流程概述
        ----------
        1. 查询 MySQL 表并转换为 DataFrame。
        2. 筛选有效来源并排除已在 NSFC 表中的记录。
        3. 聚合分段信息为 JSON 列表。
        4. 合并标签信息为字典列。
        5. 标准化日期字段。
        6. 填补缺失标签。
        7. 输出 Pathway 调试表。

        :return: pw.Table
            处理完成的结构化信息表。
        """
        # 读取源表
        info_list = pd.DataFrame(ResourceInformationList.select().dicts())
        source_dict = pd.DataFrame(ResourceSourceDict.select().dicts())
        info_section = pd.DataFrame(ResourceInformationSectionList.select(
            ResourceInformationSectionList.section_id,
            ResourceInformationSectionList.information_id,
            ResourceInformationSectionList.section_order,
            ResourceInformationSectionList.section_attr,
            ResourceInformationSectionList.title_level,
            ResourceInformationSectionList.marc_code,
            ResourceInformationSectionList.src_text,
            ResourceInformationSectionList.dst_text,
            ResourceInformationSectionList.media_info,
            ResourceInformationSectionList.md5_encode

        ).dicts())
        info_tags_relation = pd.DataFrame(ResourceInformationTagsRelation.select(
            ResourceInformationTagsRelation.information_id,
            ResourceInformationTagsRelation.tag_code,
            ResourceInformationTagsRelation.tag_value
        ).dicts())
        info_attachment = pd.DataFrame(ResourceInformationAttachmentList.select().dicts())
        nsfc_list = pd.DataFrame(NsfcInfoList.select().dicts())

        # 标签字段类型统一，确保 join 正确
        info_tags_relation["information_id"] = info_tags_relation["information_id"].astype(str)
        info_tags_relation["tag_code"] = info_tags_relation["tag_code"].astype(str)
        info_tags_relation["tag_value"] = info_tags_relation["tag_value"].apply(lambda x: eval(x))

        # 来源校验：仅保留在 source_dict 中存在的来源
        info_list = info_list[info_list["source_id"].isin(source_dict["source_id"])]

        # 排除已在 NSFC 表中存在的信息
        if len(nsfc_list) > 0:
            info_list = info_list[~info_list["information_id"].isin(nsfc_list["information_id"])]

        # 将 info_section 按 information_id 聚合为列表
        info_section_grouped = info_section.groupby("information_id").apply(
            lambda x: x.to_dict(orient="records")  # 将每个分组转成字典列表
        ).to_dict()  # 转成字典，方便 map 使用

        # 将聚合后的列表映射回 info_list
        info_list["info_section"] = info_list["information_id"].map(info_section_grouped)

        # 对没有分段信息的填充空列表
        info_list["info_section"] = info_list["info_section"].apply(lambda x: x if isinstance(x, list) else [])

        # 将 info_attachment 按 information_id 聚合为列表
        info_attachment_grouped = info_attachment.groupby("information_id").apply(
            lambda x: x.to_dict(orient="records")  # 将每个分组转成字典列表
        ).to_dict()  # 转成字典，方便 map 使用

        # 将聚合后的列表映射回 info_list
        info_list["info_attachment"] = info_list["information_id"].map(info_attachment_grouped)

        # 对没有附件信息的填充空列表
        info_list["info_attachment"] = info_list["info_attachment"].apply(lambda x: x if isinstance(x, list) else [])

        # 标签信息聚合为字典
        merged_df = info_list.merge(
            info_tags_relation,
            how="left",
            on="information_id",
            suffixes=("", "_source"),
        )
        info_list["info_tags"] = merged_df[info_tags_relation.columns].apply(lambda row: row.to_dict(), axis=1)

        # 日期标准化
        info_list["publish_date"] = pd.to_datetime(info_list["publish_date"], errors="coerce")
        info_list["publish_date"] = info_list["publish_date"].dt.strftime("%Y-%m-%d").fillna("")

        # 填补缺失标签
        info_list["info_tags"] = info_list["info_tags"].where(pd.notna(info_list["info_tags"]), {})

        # 重置索引并输出调试文件
        info_list = info_list.reset_index(drop=True)
        # 转换为 Pathway 表对象
        input_data = pw.debug.table_from_pandas(info_list)
        return input_data

    def write(self, transform_data):
        """
        将 transform 阶段产出的 Pathway 表数据写入目标。

        默认写入 NSFC 数据库，也可切换 PrintWriter 用于调试。

        :param transform_data: pw.Table
            已完成 transform 的 Pathway 表对象。
        :return: None
        """
        # 调试输出至控制台
        # pw.io.python.write(transform_data, PrintWriter())
        #
        # 实际写入 NSFC 库
        pw.io.python.write(transform_data, NsfcToNsfcWrite())
